import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.ProvinceTransactionBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DwsProvinceTransactionWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsProvinceTransactionWindow().start(
                13432,
                4,
                "dws_province_province_transaction_window",
                Constant.DWD_TRANS_PAY
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSON.parseObject(jsonStr);
                    }
                }
        );

        //以唯一主键分组去重
        KeyedStream<JSONObject, String> idKeyedDs = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //定时器去重 只获取最后一个
        SingleOutputStreamOperator<JSONObject> distinctDs = idKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> jsonObjectValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<JSONObject> stateProperties =
                                new ValueStateDescriptor<JSONObject>("jsonObjectValueState", JSONObject.class);
                        jsonObjectValueState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        JSONObject lastJsonObj = jsonObjectValueState.value();
                        if (lastJsonObj == null) {
                            jsonObjectValueState.update(jsonObj);

                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = jsonObjectValueState.value();
                        out.collect(jsonObj);
                    }
                }
        );

        KeyedStream<JSONObject, String> orderIdKeyedDs =
                distinctDs.keyBy(jsonObj -> jsonObj.getString("order_id"));

        SingleOutputStreamOperator<JSONObject> processDs = orderIdKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> lastOrderIdState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<String>("lastOrderIdState", String.class);
                        lastOrderIdState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = new JSONObject();
                        Long ts = jsonObj.getLong("ts") * 1000;
                        String provinceId = jsonObj.getString("province_id");
                        String amount = jsonObj.getString("final_amount");
                        String curOrderId = jsonObj.getString("order_id");
                        //上一个orderid，因为以orderId分组，如果第二次orderid等于上一次orderid说明是同一个订单
                        String lastOrderId = lastOrderIdState.value();

                        jsonObject.put("amount", amount);
                        jsonObject.put("orderUser", 1);
                        jsonObject.put("orderCount", 1);
                        jsonObject.put("provinceId",provinceId);
                        jsonObject.put("ts", ts);

                        if (lastOrderId != null && lastOrderId.equals(curOrderId)) {
                            jsonObject.put("orderUser", 0);
                            jsonObject.put("orderCount", 0);
                        }


                        out.collect(jsonObject);

                    }
                }
        );

        //processDs.print();
        //1> {"orderUser":1,"amount":"200.0","orderCount":1,"provinceId":"31","ts":1731718958000}
        //3> {"orderUser":1,"amount":"200.0","orderCount":1,"provinceId":"3","ts":1731718958000}



        SingleOutputStreamOperator<ProvinceTransactionBean> ptBean = processDs.map(
                new MapFunction<JSONObject, ProvinceTransactionBean>() {
                    @Override
                    public ProvinceTransactionBean map(JSONObject jsonObj) throws Exception {
                        ProvinceTransactionBean transactionBean = ProvinceTransactionBean.builder()
                                .stt("")
                                .edt("")
                                .curDate("")
                                .province(jsonObj.getString("provinceId"))
                                .orderUser(jsonObj.getLong("orderUser"))
                                .orderCount(jsonObj.getLong("orderCount"))
                                .amount(jsonObj.getDouble("amount"))
                                .ts(jsonObj.getLong("ts"))
                                .build();
                        return transactionBean;
                    }
                }
        );

        //ptBean.print();



        SingleOutputStreamOperator<ProvinceTransactionBean> withWatermarkDs = ptBean.assignTimestampsAndWatermarks(
                WatermarkStrategy.<ProvinceTransactionBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<ProvinceTransactionBean>() {
                                    @Override
                                    public long extractTimestamp(ProvinceTransactionBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        KeyedStream<ProvinceTransactionBean, String> provinceKeyedDs = withWatermarkDs.keyBy(ProvinceTransactionBean::getProvince);

        WindowedStream<ProvinceTransactionBean, String, TimeWindow> windowDs = provinceKeyedDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<ProvinceTransactionBean> reduceDs = windowDs.reduce(
                new ReduceFunction<ProvinceTransactionBean>() {
                    //4> ProvinceTransactionBean(stt=, edt=, cur_date=, province=19, amount=200.0, orderUser=1, orderCount=1, ts=1731720087000)
                    @Override
                    public ProvinceTransactionBean reduce(ProvinceTransactionBean value1, ProvinceTransactionBean value2) throws Exception {
                        value1.setAmount(value1.getAmount() + value2.getAmount());
                        value1.setOrderUser(value1.getOrderUser() + value2.getOrderUser());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        return value1;
                    }
                },
                new ProcessWindowFunction<ProvinceTransactionBean, ProvinceTransactionBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, ProcessWindowFunction<ProvinceTransactionBean, ProvinceTransactionBean, String, TimeWindow>.Context context, Iterable<ProvinceTransactionBean> elements, Collector<ProvinceTransactionBean> out) throws Exception {
                        TimeWindow window = context.window();
                        ProvinceTransactionBean provinceTransactionBean = elements.iterator().next();
                        ProvinceTransactionBean bean = ProvinceTransactionBean.builder()
                                .stt(DateFormatUtil.tsToDateTime(window.getStart()))
                                .edt(DateFormatUtil.tsToDateTime(window.getEnd()))
                                .curDate(DateFormatUtil.tsToDate(window.getStart()))
                                .province(provinceTransactionBean.getProvince())
                                .orderUser(provinceTransactionBean.getOrderUser())
                                .orderCount(provinceTransactionBean.getOrderCount())
                                .amount(provinceTransactionBean.getAmount())
                                .ts(provinceTransactionBean.getTs())
                                .build();
                        out.collect(bean);
                    }
                }
        );

        SingleOutputStreamOperator<ProvinceTransactionBean> resultDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<ProvinceTransactionBean>() {
                    @Override
                    protected void addDim(ProvinceTransactionBean obj, JSONObject dimJsonObj) {
                        obj.setProvince(dimJsonObj.getString("name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_province";
                    }

                    @Override
                    public String getRowKey(ProvinceTransactionBean obj) {
                        return obj.getProvince();
                    }
                },
                60, TimeUnit.SECONDS
        );

        resultDs.print();

        SingleOutputStreamOperator<String> jsonBean = resultDs.map(new BeanToJsonStrMapFunction<>());

        jsonBean.print();

        jsonBean.sinkTo(FlinkSinkUtil.getDorisSink("dws_province_province_transaction_window"));

    }
}
